# -*- coding: utf-8 -*-
# =============================================================================
#         Desc: 使用mysqlbinlog命令对binlog进行分析找出
#               1、执行时间较长事务。
#               2、影响行数较多事务。
#               3、产生Binlog量较大的事务。
#       Author: GGA
#        Email:
#     HomePage:
#      Version: 1.0.0
#   LastChange: 2020-12-20
#      History:
# =============================================================================
import os
import sys
import subprocess
import logging
import re
import datetime

# mysqlbinlog文件路径
mysql_binlog_exe = "mysqlbinlog"
# 事务最大BINLOG长度，超过后被判定为大事务
max_tran_binlog_size = 1 * 1024 * 1024
# 事务最大的DML数量，超过后判定为大事务
max_tran_dml_count = 1000
# 事务最大的持续时间，超过则判定大事务
max_tran_run_seconds = 2
# 采集大事务的数量，超过该数量后停止采集
max_big_tran_count = 1000

logger = logging.getLogger()


class LoggerHelper(object):
    @classmethod
    def init_logger(cls):
        logger.setLevel(level=logging.DEBUG)
        logger_format = logging.Formatter("[%(asctime)s]-[%(levelname)s]: %(message)s")  # output format
        sh = logging.StreamHandler(stream=sys.stdout)  # output to standard output
        sh.setFormatter(logger_format)
        logger.addHandler(sh)


class TransactionInfo(object):
    DML_FLAGS = ["### INSERT INTO", "### UPDATE", "### DELETE FROM"]

    def __init__(self):
        self.start_post = 0
        self.stop_post = 0
        self.start_time = None
        self.stop_time = None
        self.dml_map = dict()

    def get_dml_count(self):
        dml_count = 0
        for key_item in self.dml_map.keys():
            key_value = self.dml_map[key_item]
            dml_count += key_value
        return dml_count

    def get_binlog_size(self):
        return self.stop_post - self.start_post

    def get_tran_seconds(self):
        if (self.stop_time is not None) and (self.start_time is not None):
            tran_seconds = (self.stop_time - self.start_time).total_seconds()
            return tran_seconds
        else:
            return 0

    def is_big_tran(self):
        if self.get_binlog_size() >= max_tran_binlog_size:
            return True
        if self.get_dml_count() >= max_tran_dml_count:
            return True
        if self.get_tran_seconds() > max_tran_run_seconds:
            return True
        return False

    def show_trans(self):
        print("=" * 30)
        print("事务开始位点:{}".format(self.start_post))
        print("事务结束位点:{}".format(self.stop_post))
        print("事务开始时间:{}".format(self.start_time.strftime("%Y-%m-%d %H:%M:%S")))
        print("事务结束时间:{}".format(self.stop_time.strftime("%Y-%m-%d %H:%M:%S")))
        print("事务BINLOG长度:{}".format(self.get_binlog_size()))
        print("事务DML次数:{}".format(self.get_dml_count()))
        print("事务持续时间:{}".format(self.get_tran_seconds()))
        for key_item in self.dml_map.keys():
            key_value = self.dml_map[key_item]
            print("{}操作次数：{}".format(key_item, key_value))

    def update_dml_map(self, binlog_line):
        dml_key = ""
        for dml_flag in self.DML_FLAGS:
            if binlog_line.find(dml_flag) >= 0:
                dml_key = binlog_line.strip()
                break
        if dml_key == "":
            return
        if dml_key in self.dml_map.keys():
            self.dml_map[dml_key] = self.dml_map[dml_key] + 1
        else:
            self.dml_map[dml_key] = 1


def get_binlog_datetime(binlog_line):
    if binlog_line.find("server id") < 0:
        return None
    start_index = 1
    end_index = binlog_line.index("server id", start_index)
    dt_str = binlog_line[start_index:end_index].strip().replace("#", "").strip()
    re_pattern = r'\d{6}\s{1,2}\d{1,2}:\d{2}:\d{2}'
    re_match = re.match(re_pattern, dt_str)
    if re_match:
        dt_str = "20" + dt_str
    dt_tmp = datetime.datetime.strptime(dt_str, "%Y%m%d %H:%M:%S")
    return dt_tmp


def is_qmq_only(table_ops: list):
    has_qmq_insert = False
    has_other_table = False
    for table_op in table_ops:
        if table_op == "### INSERT INTO `qmq_produce`.`qmq_msg_queue`":
            has_qmq_insert = True
        if str(table_op).find("`qmq_produce`.`qmq_msg_queue`") < 0:
            has_other_table = True
    if has_qmq_insert and not has_other_table:
        return True
    else:
        return False


def is_qmq_tran(table_ops: list):
    has_qmq_insert = False
    has_other_table = False
    for table_op in table_ops:
        if table_op == "### INSERT INTO `qmq_produce`.`qmq_msg_queue`":
            has_qmq_insert = True
        if str(table_op).find("`qmq_produce`.`qmq_msg_queue`") < 0:
            has_other_table = True
    if has_qmq_insert and has_other_table:
        return True
    else:
        return False


def add_qmq_subject_count(subject_counter: dict, subject_title: str):
    if subject_title not in subject_counter.keys():
        subject_counter[subject_title] = 0
    subject_counter[subject_title] += 1


def get_subject_title(subject_message_line: str):
    return str(subject_message_line.split(",")[1]).split(":")[1]


def show_qmq_counter_info(qmq_counter):
    qmq_subjects = sorted(qmq_counter.items(), key=lambda x: x[1], reverse=True)
    qmq_messages = list()
    qmq_messages.append("")
    qmq_messages.append("*" * 50)
    qmq_messages.append("%-15s       %s" % ("操作次数", "操作主题"))
    for qmq_subject in qmq_subjects:
        qmq_messages.append("%-15s       %s" % (qmq_subject[1], qmq_subject[0]))
    qmq_messages.append("*" * 50)
    logger.info("\n".join(qmq_messages))


def get_qmq_summary(dump_sql_path):
    logger.info("开始检查文件：{}".format(dump_sql_path))
    cmd = """ {} -v --base64-output=DECODE-ROWS '{}' | egrep '^BEGIN|^COMMIT|^### INSERT|^### UPDATE|^### DELETE|^###   @7=|^# at|server id' """
    cmd = cmd.format(
        mysql_binlog_exe, dump_sql_path
    )
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    last_log_pos = 0
    table_ops = []
    line_num = 0
    qmq_only_counter = dict()
    qmq_tran_counter = dict()
    qmq_subject = ""
    start_time = None
    stop_time = None
    while True:
        file_bytes = p.stdout.readline()
        if not file_bytes:
            break
        file_line = file_bytes.decode().strip()
        line_num += 1
        if line_num % 10000 == 0:
            logger.info("处理记录行：{},处理位点：{},处理记录信息：{}".format(line_num, last_log_pos, file_line))
        if file_line.startswith("###   @7=") and file_line.find("messageId") > 0 and file_line.find("subject") > 0:
            qmq_subject = get_subject_title(file_line)
            continue
        if file_line.find("server id") > 0:
            current_time = get_binlog_datetime(file_line)
            if start_time is None:
                start_time = current_time
            stop_time = current_time
            continue
        if file_line.startswith("# at"):
            last_log_pos = int(str(file_line).replace("# at", "").strip())
            continue
        if file_line.startswith("BEGIN") > 0:
            # 事务开始
            table_ops = []
            continue
        if file_line.startswith("COMMIT/*!*/;") > 0:
            if is_qmq_only(table_ops=table_ops):
                add_qmq_subject_count(qmq_only_counter, qmq_subject)
            if is_qmq_tran(table_ops=table_ops):
                add_qmq_subject_count(qmq_tran_counter, qmq_subject)
            continue
        if file_line.startswith("### INSERT") > 0 \
                or file_line.startswith("### UPDATE") > 0 \
                or file_line.startswith("### DELETE") > 0:
            table_ops.append(file_line.strip())
            continue
    logger.info("检查文件完成：{}".format(dump_sql_path))
    logger.info("检查开始时间：{}".format(start_time.strftime("%Y-%m-%d %H:%M:%S")))
    logger.info("检查结束时间：{}".format(stop_time.strftime("%Y-%m-%d %H:%M:%S")))
    logger.info("下面是单独插入`qmq_produce`.`qmq_msg_queue`表的主题")
    show_qmq_counter_info(qmq_only_counter)
    logger.info("下面是混合插入`qmq_produce`.`qmq_msg_queue`表的主题")
    show_qmq_counter_info(qmq_tran_counter)
    logger.info("检查完成")


def show_usage():
    logger.info("解析mysql binlog文件分析")
    logger.info("使用语法: python mysql_binlog_qmq_checker.py mysql-bin.10001")


def main(args):
    if len(args) != 2:
        logger.info("传入参数错误,请重试！")
        show_usage()
        return
    if not os.path.exists(args[1]):
        logger.info("指定的文件不存在,请重试！")
        show_usage()
        return
    get_qmq_summary(dump_sql_path=args[1])


if __name__ == '__main__':
    LoggerHelper.init_logger()
    main(sys.argv)
